package com.tzu.seq.bean;

import com.alipay.sofa.jraft.util.Bits;
import com.alipay.sofa.jraft.util.BytesUtil;
import com.google.common.collect.Lists;
import io.vertx.core.buffer.Buffer;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang.ArrayUtils;
import thirdpart.bean.CmdPack;
import thirdpart.fetchsurv.IFetchService;
import thirdpart.order.OrderCmd;
import thirdpart.order.OrderDirection;

import java.util.List;
import java.util.Map;
import java.util.TimerTask;

/**
 * @Author: Tzu
 * @Date: 2020/12/9 18:24
 * @Version: 1.0
 * @Description:
 */

@Log4j2
@RequiredArgsConstructor
public class FetchTask extends TimerTask {

    @NonNull
    private SeqConfig config;

    @Override
    public void run() {
        //1、判断当前node节点不是leader：leader节点负责写入订单数据
        if (!config.getNode().isLeader()){
            return;
        }

        //2、获取leader节点排队机与网关的连接记录
        Map<String, IFetchService> fetchServiceMap = config.getFetchServiceMap();
        if (MapUtils.isEmpty(fetchServiceMap)){
            //连接记录为空 结束程序
            return;
        }

        //3、通过记录 遍历网关 拉取订单数据
        List<OrderCmd> cmds = collectAllOrders(fetchServiceMap);
        //验空
        if (CollectionUtils.isEmpty(cmds)){
            return;
        }
        //在控制台打印从网关拉取的订单信息
        log.info("recv order :[{}]",cmds);

        //4、对从网关获取的订单数据进行排序
        //定序规则(自定义)：时间优先-->价格优先-->量优先
        cmds.sort((o1,o2) -> {
            //比较创建订单时间
            int res = compareTime(o1,o2);
            if (res != 0){
                return res;
            }
            //创建订单时间相同则比较价格
            res = comparePrice(o1,o2);
            if (res != 0){
                return res;
            }
            //创建订单时间、价格相同，比较数量
            res = compareVolume(o1,o2);

            return res;
        });

        //5、存储到KV store, 发送到撮合核心
        try {
            //① 生成 packetno
            long packetNo = getPacketNoFromStore();
            //② 入库
            CmdPack pack = new CmdPack(packetNo, cmds);
            byte[] serialize = config.getCodec().serialize(pack);
            insertToKvStore(packetNo,serialize);
            //③ 更新 packNo+1
            updatePacketNoInStore(packetNo + 1);
            //④ 发送
            config.getMulticastSender().send(
                    Buffer.buffer(serialize),
                    config.getMulticastPort(),
                    config.getMulticastIp(),
                    null
            );

        }catch (Exception e){
            log.error("encode cmd packet error");
        }
    }

    /////////////////////////更新PacketNo+1并存入kvstore//////////////////////////////////////////

    private void updatePacketNoInStore(long packetNo){
        final byte[] bytes = new byte[8];
        Bits.putLong(bytes,0,packetNo);
        config.getNode().getRheaKVStore().put(PACKET_NO_KEY,bytes);
    }

    /////////////////////////保存数据到kvstore//////////////////////////////////////////

    private void insertToKvStore(long packetNo, byte[] serialize){
        byte[] key = new byte[8];
        Bits.putLong(key,0,packetNo);
        //put是异步的，but是同步操作
        config.getNode().getRheaKVStore().put(key,serialize);
    }

    /////////////////////////获取packetNo//////////////////////////////////////////
    //kv store的键都是byte[]类型
    private static final byte[] PACKET_NO_KEY = BytesUtil.writeUtf8("seq_packet_no");

    private long getPacketNoFromStore(){
        //从kv store获取packetNo
        final byte[] bPacketNo = config.getNode().getRheaKVStore().bGet(PACKET_NO_KEY);
        //验空
        long packetNo = 0;
        if (ArrayUtils.isNotEmpty(bPacketNo)){
            packetNo = Bits.getLong(bPacketNo,0);
        }

        return packetNo;
    }

    /////////////////////////定序规则//////////////////////////////////////////

    private int compareVolume(OrderCmd o1, OrderCmd o2) {
        if (o1.volume > o2.volume){
            //购买数量大的订单排在前面
            return -1;
        }else if (o1.volume < o2.volume){
            //购买数量小的订单排在后面
            return 1;
        }else {
            //o1、o2购买数量相同，无法比较
            return 0;
        }
    }

    private int comparePrice(OrderCmd o1, OrderCmd o2) {
        if (o1.direction == o2.direction){
            //o1,o2订单买卖方向相同才能比较
            if (o1.price > o2.price){
                //若o1、o2为买订单，价格大的排在前面
                //若o1、o2为卖订单，价格大的排在后面
                return o1.direction == OrderDirection.BUY ? -1 : 1;
            }else if (o1.price < o2.price){
                return o1.direction == OrderDirection.BUY ? 1 : -1;
            }else {
                //o1,o2价格相同，通过比较价格无法定序
                return 0;
            }
        }else {
            //o1,o2订单买卖方向不同，这两个订单谁前谁后都可以
            return 0;
        }
    }

    private int compareTime(OrderCmd o1, OrderCmd o2) {
        if (o1.timestamp > o2.timestamp){
            //o1排在o2后面
            return 1;
        }else if (o1.timestamp < o2.timestamp){
            //o1排在o2前面
            return -1;
        }else {
            //创建订单时间相同，通过比较订单创建时间无法定序
            return 0;
        }
    }

   ////////////////////从网关收集订单数据///////////////////////////////////

    private List<OrderCmd> collectAllOrders(Map<String,IFetchService> fetchServiceMap){
        //不推荐这种写法：性能不可控 调试不方便 可读性差
//        List<OrderCmd> res = fetchServiceMap.values().stream()
//                .map(t -> t.fetchData())
//                .filter(msg -> CollectionUtils.isEmpty(msg))
//                .flatMap(List::stream)
//                .collect(Collectors.toList());

        //推荐
        List<OrderCmd> msgs = Lists.newArrayList();
        //遍历连接记录
        fetchServiceMap.values().forEach(t -> {
            //通过IFetchService对象的fetchData()方法到对应网关拉取订单数据
            List<OrderCmd> orderCmds = t.fetchData();
            //验空
            if (CollectionUtils.isNotEmpty(orderCmds)){
                msgs.addAll(orderCmds);
            }
        });

        return msgs;
    }
}
